package cn.juque.redoctopus.listener;

import cn.juque.common.base.MqMessageDTO;
import cn.juque.common.disruptor.DisruptorHandler;
import cn.juque.redoctopus.api.dto.LogInfoDTO;
import cn.juque.redoctopus.bo.disruptor.LogBO;
import cn.juque.redoctopus.disruptor.event.LogEventServiceImpl;
import com.rabbitmq.client.Channel;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author nuoka
 * @version 1.0.0
 * <li>IntelliJ IDEA</li>
 * <li>cn.juque.redoctopus.listener</li>
 * @date 2020/9/10 19:15
 **/
@Component
@Slf4j
@RabbitListener(queues = {"octopus.log1","octopus.log2"})
public class OctopusLogListener {

    private static final Integer MSG_TYPE = 1;

    @Resource
    private DisruptorHandler disruptorHandler;

    @RabbitHandler
    public void process(Channel channel, Message message, MqMessageDTO<LogInfoDTO> mqMessage) {
        try {
            if (MSG_TYPE.equals(mqMessage.getType())) {
                log.info("mq message:{}", mqMessage);

                LogInfoDTO logInfoBO = mqMessage.getPayload();
                if(null == logInfoBO) {
                    return ;
                }
                LogBO logBO = this.toLogBO(logInfoBO);
                logBO.setBusinessEventService(LogEventServiceImpl.BEAN_NAME);
                this.disruptorHandler.publish(logBO);
                //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了
                // 否则消息服务器以为这条消息没处理掉 后续还会在发
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }
        } catch (Exception e) {
            log.error("user-rel-org mq error", e);
        }
    }

    private LogBO toLogBO(LogInfoDTO bo) {
        LogBO logBO = new LogBO();
        logBO.setTimestamp(bo.getTimestamp());
        logBO.setServiceNo(bo.getServiceNo());
        logBO.setServiceName(bo.getServiceName());
        logBO.setLevel(bo.getLevel());
        logBO.setMessage(bo.getMessage());
        logBO.setThrowable(bo.getThrowable());
        logBO.setIp(bo.getIp());
        logBO.setThreadId(bo.getThreadId());
        return logBO;
    }
}
